package com.yanggu.bigdata.realtime.app.function

import cn.hutool.core.collection.CollUtil
import com.alibaba.fastjson.JSONObject
import com.yanggu.bigdata.realtime.app.OutputTagObject.{displayOutputTag, startLogOutputTag}
import org.apache.flink.streaming.api.functions.ProcessFunction
import org.apache.flink.util.Collector

/**
 * ods层行为日志数据进行分流输出到侧输出流
 */
class OdsBaseLogToDwdSideOutputProcessFunction extends ProcessFunction[JSONObject, String] {

  override def processElement(value: JSONObject, ctx: ProcessFunction[JSONObject, String]#Context, out: Collector[String]): Unit = {
    val startJsonObject = value.getJSONObject("start")
    //包含start, 说明是启动日志
    if (CollUtil.isNotEmpty(startJsonObject)) {
      //启动日志直接输出到测输出流
      ctx.output(startLogOutputTag, value.toJSONString)
    } else {
      //如果是页面数据, 直接输出到主流
      out.collect(value.toJSONString)
      //如果不是启动日志, 继续判断是否为曝光数据
      val displays = value.getJSONArray("displays")
      if (CollUtil.isNotEmpty(displays)) {
        val pageId = value.getJSONObject("page").getString("page_id")
        for (index <- 1 until displays.size()) {
          val displayJson = displays.getJSONObject(index)
          displayJson.put("page_id", pageId)
          ctx.output(displayOutputTag, displayJson.toJSONString)
        }
      }
    }
  }

}
